/*
    Copyright (C) 2021 Huawei Device Co., Ltd.
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.danikula.videocache.onefileStrategy;

import com.danikula.videocache.*;
import com.danikula.videocache.util.LogUtil;
import com.danikula.videocache.util.ProxyCacheUtils;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;

import static com.danikula.videocache.util.ProxyCacheUtils.DEFAULT_BUFFER_SIZE;

/**
 * 分节点缓存代理
 */
public class OneFileProxyCache extends ProxyCache {
    /**
     * 日志打印工具
     */
    private static final String TAG = "OneFileProxyCache";

    private final HttpUrlSource source;
    private final OneFileCache cache;


    public OneFileProxyCache(HttpUrlSource source, OneFileCache cache) {
        super(source, cache);
        this.cache = cache;
        this.source = source;
    }


    @Override
    public void processRequest(GetRequest request, Socket socket) throws IOException, InterruptedException,
            ProxyCacheException {
        OutputStream out = new BufferedOutputStream(socket.getOutputStream());
        String responseHeaders = newResponseHeaders(request);
        out.write(responseHeaders.getBytes("UTF-8"));
        long offset = request.rangeOffset;
        if (source.length() > 0) {
            cache.setFileLength(source.length());
        }
        if (cache.isTempFile(cache.file)) {
            long[] downloadArea = cache.getDownloadArea(offset);
            cache.initDownLoadData();
            cache.setCacheStart(downloadArea[0] > 0 ? downloadArea[0] : 0);
            cache.setCacheEnd(downloadArea[1]);
            responseWithCache(out, offset);
        } else {
            cache.setCacheStart(0);
            cache.setCacheEnd(cache.file.length());
            // 整个文件缓存完毕重命名之后还是被代理 直接读取文件即可
            onCachePercentsAvailableChanged(100);
            responseWithFile(out, offset);
        }
    }

    @Override
    public int read(byte[] buffer, long offset, int length) throws ProxyCacheException {
        try {
            ProxyCacheUtils.assertBuffer(buffer, offset, length);
            while (!cache.isCompleted() && !cache.isCacheEnough(offset, length) && !stopped) {
                readSourceAsync();
                waitForSourceData();
                checkReadSourceErrorsCount();
            }
            int read = cache.find(buffer, offset, length);
            if (cache.isCompleted() && percentsAvailable != 100) {
                percentsAvailable = 100;
                onCachePercentsAvailableChanged(100);
            }
            return read;
        } catch (IOException e) {
            LogUtil.error(this.getClass().getSimpleName(), e.toString());
        }
        return -1;
    }

    @Override
    protected Runnable getRunable() {
        return new OneFileSourceRunnable();
    }

    @Override
    protected void onCachePercentsAvailableChanged(int percents) {
        if (listener != null) {
            listener.onCachePartAvailable(cache.file, source.getUrl(), cache.getDownloadArea(), cache.getFileLength());
        }
    }

    private class OneFileSourceRunnable implements Runnable {
        long sourceAvailable = -1;
        long offset = 0;

        OneFileSourceRunnable() {
            // do nothing
        }

        @Override
        public void run() {
            readSource();
        }

        private void readSource() {
            try {
                offset = cache.getDownLoadStart();
                if (offset == cache.getFileLength() && offset > 0) {
                    onError(new ProxyCacheException("下载完毕了"));
                    return;
                }
                if (offset >= cache.getDownLoadEnd()) {
                    return;
                }
                source.open(offset);
                sourceAvailable = source.length();
                cache.setFileLength((int) sourceAvailable);
                if (offset == sourceAvailable) {
                    onError(new ProxyCacheException("下载完毕了"));
                    return;
                }
                cache.setStop(false);
                writeData();
                isDownLoadFinish();
                onSourceRead();
            } catch (ProxyCacheException | IOException e) {
                readSourceErrorsCount.incrementAndGet();
                onError(e);
            } finally {
                cache.setStop(true);
                closeSource();
                if (sourceAvailable > 0) {
                    notifyNewCacheDataAvailable(cache.getTotalCacheLength(), sourceAvailable);
                }
            }
        }

        private void writeData() throws ProxyCacheException, IOException {
            byte[] buffer = new byte[ProxyCacheUtils.DEFAULT_BUFFER_SIZE];
            int readBytes;
            while ((readBytes = source.read(buffer)) != -1) {
                synchronized (stopLock) {
                    if (isStopped()) {
                        cache.setStop(true);
                        return;
                    }
                    cache.insert(buffer, readBytes, offset);
                    offset += readBytes;
                    // 数据下载到其他已缓存区域的交接位置之后即不再下载 在insert之后return 允许多写部分数据覆盖交接位置 防止出现断点
                    if (offset >= cache.getDownLoadEnd() || offset >= sourceAvailable) {
                        cache.setStop(true);
                        cache.mergeWriteArea(offset);
                        break;
                    }
                }
                notifyNewCacheDataAvailable(cache.getTotalCacheLength(), sourceAvailable);
            }
        }
    }


    /**
     * 视频文件是否下载完毕
     *
     * @throws ProxyCacheException 错误信息
     */
    private void isDownLoadFinish() throws ProxyCacheException {
        synchronized (stopLock) {
            long totalCacheLength = cache.getTotalCacheLength();
            if (totalCacheLength == source.length()) {
                cache.complete();
            }
        }
    }

    @Override
    public void onSourceRead() {
        int percents = (int) ((float) cache.getTotalCacheLength() * 100 / cache.getFileLength());
        onCachePercentsAvailableChanged(percents);
    }

    @Override
    public void doComplete() throws ProxyCacheException {
        // 边写边读的时候由于都要使用 dataFile.seek
        // 两个线程一起进行如果卡顿的话会导致指针错乱数据错误
        // 所以使用 synchronized (file)来同步操作 当次读或者写完毕之后释放锁
        synchronized (stopLock) {
            LogUtil.debug(TAG, "Shutdown proxy for " + source.toString());
            tryComplete();
            closeSource();
        }
    }

    @Override
    protected void tryComplete() throws ProxyCacheException {
        stopped = true;
        cache.setStop(true);
        cache.mergeWriteArea(-1);
        long totalCacheLength = cache.getTotalCacheLength();
        int percents = (int) ((float) totalCacheLength * 100 / cache.getFileLength());
        onCachePercentsAvailableChanged(percents);
        if (totalCacheLength == source.length()) {
            cache.complete();
        }
    }

}
